package org.quxiao.hbase.test.addData;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.hfile.Compression.Algorithm;
import org.apache.hadoop.hbase.regionserver.StoreFile.BloomType;

public class AddAssetData {


	private static Configuration conf = null;
	private static HBaseAdmin admin = null;
	
	static
	{
		conf = HBaseConfiguration.create();
		try
		{
			admin = new HBaseAdmin(conf);
		}
		catch (ZooKeeperConnectionException zooEx)
		{
			System.err.println("can not connect zookeeper!");
			System.exit(-1);
		}
		catch (MasterNotRunningException masterNotRunEx)
		{
			System.err.println("hbase master isn't running!");
			System.exit(-1);
		}
	}
	
	private static void addAssetMeta(String filePath, String tableName)
	{
		BufferedReader br = null;
		String line = null;
		AssetMetaData meta = null;
		long writeBufferSize = 10 * 1024 * 1024;
		
		try
		{
			HTableDescriptor tableDescriptor = new HTableDescriptor(tableName.getBytes());
			
			HColumnDescriptor metaDataFamily = new HColumnDescriptor("MetaData");
			metaDataFamily.setInMemory(true);
			metaDataFamily.setCompressionType(Algorithm.GZ);
			metaDataFamily.setBloomFilterType(BloomType.ROWCOL);
			
			HColumnDescriptor eventFamily = new HColumnDescriptor("Event");
			eventFamily.setInMemory(true);
			eventFamily.setCompressionType(Algorithm.GZ);
			eventFamily.setBloomFilterType(BloomType.ROWCOL);
			
			tableDescriptor.addFamily(metaDataFamily);
			tableDescriptor.addFamily(eventFamily);
			
			if ( ! admin.tableExists(tableName) )
			{
				admin.createTable(tableDescriptor);
			}
			
			HTable table = new HTable(conf, tableName);
			table.setAutoFlush(false);
			table.setWriteBufferSize(writeBufferSize);
			
			List<Put> putList = new ArrayList<Put>();
			br = new BufferedReader(new FileReader( new File(filePath)));
			while ( (line = br.readLine()) != null )
			{
				meta = parseLineToMeta(line);
				if ( meta != null )
				{
					Put putRow = new Put(Integer.valueOf(meta.assetId).toString().getBytes());
					putRow.setWriteToWAL(false);
					putRow.add("MetaData".getBytes(), "vendor".getBytes(), meta.vendor.getBytes());
					putRow.add("MetaData".getBytes(), "type".getBytes(), meta.type.getBytes());
					putRow.add("MetaData".getBytes(), "price".getBytes(), Integer.valueOf(meta.price).toString().getBytes());
					putRow.add("MetaData".getBytes(), "tagId".getBytes(), Integer.valueOf(meta.tagId).toString().getBytes());
					putList.add(putRow);
					if ( putList.size() == 1000 )
					{
						table.put(putList);
						putList.clear();
					}
//					System.out.println(meta.assetId);
				}
				if ( putList.size() != 0 )
				{
					table.put(putList);
					putList.clear();
				}
			}
		}
		catch (IOException ioEx)
		{
			ioEx.printStackTrace();
		}
	}
	
	private static void addAssetEvent(String filePath, String tableName)
	{
		BufferedReader br = null;
		String line = null;
		AssetEventData event = null;
		long writeBufferSize = 10 * 1024 * 1024;
		
		try
		{
			HTableDescriptor tableDescriptor = new HTableDescriptor(tableName.getBytes());
			
			HColumnDescriptor metaDataFamily = new HColumnDescriptor("MetaData");
			metaDataFamily.setInMemory(true);
			metaDataFamily.setCompressionType(Algorithm.GZ);
			metaDataFamily.setBloomFilterType(BloomType.ROWCOL);
			
			HColumnDescriptor eventFamily = new HColumnDescriptor("Event");
			eventFamily.setInMemory(true);
			eventFamily.setCompressionType(Algorithm.GZ);
			eventFamily.setBloomFilterType(BloomType.ROWCOL);
			
			tableDescriptor.addFamily(metaDataFamily);
			tableDescriptor.addFamily(eventFamily);
			
			if ( !admin.tableExists(tableName) )
			{
				admin.createTable(tableDescriptor);
			}
			
			HTable table = new HTable(conf, tableName);
			table.setAutoFlush(false);
			table.setWriteBufferSize(writeBufferSize);
			
			List<Put> putList = new ArrayList<Put>();
			br = new BufferedReader(new FileReader( new File(filePath)));
			while ( (line = br.readLine()) != null )
			{
				event = parseLineToEvent(line);
				if ( event != null )
				{
					Put putRow = new Put(Integer.valueOf(event.assetId).toString().getBytes());
					putRow.setWriteToWAL(false);
					putRow.add("Event".getBytes(), "level".getBytes(), event.level.getBytes());
					putRow.add("Event".getBytes(), "info".getBytes(), event.info.getBytes());
					putList.add(putRow);
					if ( putList.size() == 1000 )
					{
						table.put(putList);
						putList.clear();
					}
//					System.out.println(event.assetId);
				}
				if ( putList.size() != 0 )
				{
					table.put(putList);
					putList.clear();
				}
			}
		}
		catch (IOException ioEx)
		{
			ioEx.printStackTrace();
		}
	}
	private static AssetMetaData parseLineToMeta (String line)
	{
		String[] args = line.split("\t");
		if ( args.length != 5 )
			return null;
		AssetMetaData meta = new AssetMetaData();
		meta.assetId = Integer.parseInt(args[0]);
		meta.vendor = args[1];
		meta.type = args[2];
		meta.price = Integer.parseInt(args[3]);
		meta.tagId = Integer.parseInt(args[4]);
		return meta;
	}
	
	private static AssetEventData parseLineToEvent (String line)
	{
		String[] args = line.split("\t");
		if ( args.length != 3 )
			return null;
		AssetEventData event = new AssetEventData();
		event.assetId = args[0];
		event.level = args[1];
		event.info = args[2];
		return event;
	}
	
	
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		String[] metaInputFiles = {
				"/home/ubuntu/Downloads/asset_meta_100000.out",
				"/home/ubuntu/Downloads/asset_meta_500000.out",
				"/home/ubuntu/Downloads/asset_meta_1000000.out",
				"/home/ubuntu/Downloads/asset_meta_2000000.out",
				"/home/ubuntu/Downloads/asset_meta_5000000.out",
				};
		String[] eventInputFiles = {
				"/home/ubuntu/Downloads/asset_event_100000.out",
				"/home/ubuntu/Downloads/asset_event_500000.out",
				"/home/ubuntu/Downloads/asset_event_1000000.out",
				"/home/ubuntu/Downloads/asset_event_2000000.out",
				"/home/ubuntu/Downloads/asset_event_5000000.out",
				};
		String[] tableNames = {
				"Asset_100000",
				"Asset_500000",
				"Asset_1000000",
				"Asset_2000000",
				"Asset_5000000",
		};
		assert(metaInputFiles.length == tableNames.length);
		for (int i = 0; i < metaInputFiles.length; i ++)
		{
			System.out.println("importing " + metaInputFiles[i]);
			addAssetMeta(metaInputFiles[i], tableNames[i]);
		}
		assert(eventInputFiles.length == tableNames.length);
		for (int i = 0; i < eventInputFiles.length; i ++)
		{
			System.out.println("importing " + eventInputFiles[i]);
			addAssetEvent(eventInputFiles[i], tableNames[i]);
		}
	}
}
